
package com.wsd.smartcarwasher.configuration.mqtt;

import com.wsd.smartcarwasher.service.MqttService;
import javax.annotation.Resource;
import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;
import org.json.JSONObject;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.context.annotation.Lazy;
import org.springframework.integration.annotation.ServiceActivator;
import org.springframework.integration.channel.DirectChannel;
import org.springframework.integration.core.MessageProducer;
import org.springframework.integration.mqtt.inbound.MqttPahoMessageDrivenChannelAdapter;
import org.springframework.integration.mqtt.support.DefaultPahoMessageConverter;
import org.springframework.messaging.Message;
import org.springframework.messaging.MessageChannel;
import org.springframework.messaging.MessageHandler;
import org.springframework.messaging.MessagingException;

@Configuration
public class MqttInboundConfiguration {
    private static final Log log = LogFactory.getLog(MqttInboundConfiguration.class);
    @Resource
    private MqttProperties mqttProperties;
    @Resource
    private MqttOutboundConfiguration mqttOutboundConfiguration;

    @Autowired(required = false)
    private MqttService mqttService;

    public MqttInboundConfiguration() {
    }

    @Bean
    public MessageChannel mqttInputChannel() {
        return new DirectChannel();
    }

    @Bean
    public MessageProducer inbound() {
        String[] inboundTopics = this.mqttProperties.getInbound().getTopics().split(",");
        MqttPahoMessageDrivenChannelAdapter adapter = new MqttPahoMessageDrivenChannelAdapter(this.mqttProperties.getInbound().getUrl(), this.mqttProperties.getInbound().getClientId(), this.mqttOutboundConfiguration.mqttClientFactory(), inboundTopics);
        adapter.setCompletionTimeout(5000L);
        adapter.setConverter(new DefaultPahoMessageConverter());
        adapter.setQos(new int[]{2});
        adapter.setOutputChannel(this.mqttInputChannel());
        return adapter;
    }

    @Bean
    @ServiceActivator(
            inputChannel = "mqttInputChannel"
    )
    public MessageHandler handler() {
        return new MessageHandler() {
            public void handleMessage(Message<?> message) throws MessagingException {
                String result = (String)message.getPayload();
                MqttInboundConfiguration.log.info(String.format("receive message: message[ %s ]", result));

                try {
                    MqttInboundConfiguration.this.mqttService.receiveMessage(new JSONObject(result));
                } catch (Exception var4) {
                    MqttInboundConfiguration.log.error("catch error", var4);
                }

            }
        };
    }


}
